package com.yusc.business;

import com.google.gson.Gson;
import com.yusc.pojo.Integral;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.storage.StorageLevel;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

/**
 * 5、按月统计积分兑换次数，并按兑换次数进行排序：(实时处理) --> 为了方便观察：窗口和滑动间隔都设置为30秒
 **/
public class Take05 {
    private static Gson gson = new Gson();
    private static int top = 0;
    private static int flag = 1;

    public static void main(String[] args) throws InterruptedException {
        SparkConf sparkConf = new SparkConf().setAppName("Count the number of points exchanged").setMaster("local[2]");
        JavaSparkContext jsc = new JavaSparkContext(sparkConf);

        //批次间隔是1秒
        JavaStreamingContext jStc = new JavaStreamingContext(jsc, new Duration(1000));
        JavaReceiverInputDStream<String> dStream = jStc.socketTextStream("localhost", 9999, StorageLevel.MEMORY_AND_DISK());

        //窗口长度为60秒，滑动间隔是60秒
        JavaDStream<String> window = dStream.window(new Duration(1000 * 30), new Duration(1000 * 30));

        window.map(x -> gson.fromJson(x, Integral.class))
                .mapToPair(x -> {
                    x.setExchangeCount(1);
                    return new Tuple2<>(x.getMemberNo(), x);
                })
                .reduceByKey((a, b) -> {
                    int total = a.getExchangeCount() + b.getExchangeCount();
                    a.setExchangeCount(total);
                    return a;
                })
                .map(x -> x._2)
                .transform(x -> x.sortBy(Integral::getExchangeCount, false, x.getNumPartitions()))
                .print(5);
//                .foreachRDD(x -> {
//                    x.foreachPartition(s -> {
//                        Class.forName("com.mysql.cj.jdbc.Driver");
//                        String url = "jdbc:mysql://localhost:3306/hadoop?characterEncoding=utf-8&serverTimezone" +
//                                "=Asia/Shanghai&useUnicode=true&useSSL=false&allowPublicKeyRetrieval=true";
//                        Connection conn = DriverManager.getConnection(url, "root", "123456");
//
//                        PreparedStatement ps = conn.prepareStatement("insert into `integral`" +
//                                "(member_no, ffp_tier, work_city, age, exchange_count,exchange_time,write_the_number) VALUES " +
//                                "(?,?,?,?,?,?,?)");
//
//                        while (s.hasNext()) {
//                            Integral next = s.next();
//                            ps.setString(1, next.getMemberNo());
//                            ps.setInt(2, next.getFfpTier());
//                            ps.setString(3, next.getWorkCity());
//                            ps.setInt(4, next.getAge());
//                            ps.setInt(5, next.getExchangeCount());
//                            ps.setTimestamp(6, Timestamp.valueOf(next.getExchangeTime()));
//                            ps.setInt(7,flag);
//                            ps.execute();
//                            top++;
//                            if (top == 5) { //只写入前五
//                                top = 0;
//                                flag++;  // 用于记录区分窗口滑动的次数
//                                break;
//                            }
//                        }
//                    });
//                });

        jStc.start();
        // 等待终止退出
        jStc.awaitTermination();
        jStc.stop();
    }
}
